/**
 * GeneralServiceNetManager.java
 * ************************** 
 * @date Apr 11, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package gossipServices.network;

import gossipServices.basic.nodeDescriptors.NodeDescriptor;
import gossipServices.basic.nodeDescriptors.PssNodeDescriptor;
import gossipServices.basic.nodeDescriptors.comparators.AgeInverseSorterer;

import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.util.Collections;
import java.util.Queue;
import java.util.Vector;

import networkRmi.NetworkManager;

import messages.Message;
import messages.MessageHandler;

/**
 * Description: 
 * The  GeneralServicesNetManager implements the NetworkServicesAPI interface.
 * It needs a NetworkManager as the input parameter of its constructor.
 * It's able to translate the calls of 
 * the NetworkServicesAPI interface into the functions provided
 * by the inserted NetworkManager.
 * 
 * A class of NetworkServicesAPI represents
 * the underlying layer of a service and
 * has to be able
 * to manipulate, dispatch and store messages directed to that service. 
 * The GeneralServicesNetManager inherits 
 * the duty to implement
 * the MessageHandler interface, but declares its method
 * as abstract. So all subclasses of the 
 * GeneralServicesNetManager have to provide the implementation
 * of the  extractServiceMessages(Queue<Message> queue) 
 * method with their own strategy for dispatching Message(s).
 * 
 * This system is provided by a general dispatcher, 
 * please take a look to the GeneralServicesMessageDispatcher.
 *
 */
public abstract class GeneralServicesNetManager 
	implements NetworkServicesAPI, MessageHandler {

	/**
	 * If true it will display connections errors.
	 */
	protected static boolean SHOW_SERVICE_NETMAN_ERRORS = false;
	
	/**
	 * USE_PRUNING_LISTS_METHOD controls the access to the 
	 * pruningWaitingIncoming method. It defines if the current
	 * instance has to use the pruning method, which maintains 
	 * the list to a bounded size, or not.
	 */
	protected static boolean USE_PRUNING_LISTS_METHOD = true;
	
	/**
	 * If the queue waitingIncomin break this bound limit
	 * will be pruned and the old elements are removed. 
	 */
	protected static int BOUND_WAITINGINCOMING_SIZE = 10;
	
	/**
	 * During the pruning of the waitingIncoming queue
	 * defines the age above which a NodeDescriptor is
	 * defined old and removed.
	 */
	protected static int OLD_AGE_THRESHOLD = 15;
	
	/**
	 * The list of the incoming Message(s) for the active thread
	 * (coming from a passive thread in the external node)
	 */
	protected Vector<Message> activeReceivedMessages = new Vector<Message>(100);
	
	/**
	 * The list of the incoming Message(s) for the passive thread
	 * (coming from an active thread in the external node)
	 */
	protected Vector<Message> passiveReceivedMessages = new Vector<Message>(100);
	
	/**
	 * List containing the NodeDescriptor(s) of the nodes
	 * which are going to answer to the local active thread. 
	 */
	protected Vector<NodeDescriptor> waitingIncoming = new Vector<NodeDescriptor>(100);
	
	protected NetworkManager nm;

	/**
	 * If the local active thread don't find into the
	 * "activeReceivedMessages" queue the answer which
	 * is waiting for during a "receive(NodeDescriptor node)", 
	 * it will wait a number of
	 * "nonBlockingReceiveWaitTime" milliseconds.
	 */
	protected int nonBlockingReceiveWaitTime = 10000; //msec
	
	public GeneralServicesNetManager(NetworkManager networkManager){
		nm = networkManager;
		nm.registerService(this);
	}
	
	@Override
	public String getNodeName() {
		return nm.getNodeName();
	}

	public abstract void extractIncomingMessages(Queue<Message> queue);

	@Override
	public void clearReceivedMessages() {
		activeReceivedMessages.clear();
		passiveReceivedMessages.clear();
		waitingIncoming.clear();
	}

	protected void enableNetManErrors(){
		SHOW_SERVICE_NETMAN_ERRORS = true;
	}
	
	protected void disableNetManErrors(){
		SHOW_SERVICE_NETMAN_ERRORS = false;
	}
	
	@Override
	public void send(Message msg, NodeDescriptor node) {
		try{
			nm.send(msg, node.getNodeAddress());
		}catch(RemoteException e){
			if(SHOW_SERVICE_NETMAN_ERRORS)
				System.err.println(this.toString()+" "+
						" problem sending "+ msg +" to "+msg.getDest()+
						" thrown a RemoteException");
		} catch (NotBoundException e) {
			if(SHOW_SERVICE_NETMAN_ERRORS)
				System.err.println(this.toString()+
						" problem sending "+ msg +" to "+msg.getDest()+
						" exception is "+ e);
		}
	}

	
	@Override
	public Message receive(NodeDescriptor node) throws InterruptedException {
		Message ret = null;
		
		synchronized (activeReceivedMessages) {
			if(!activeReceivedMessages.isEmpty()){
				ret = scanForMessage();
			}
		}
		
		if(ret == null){
			synchronized (activeReceivedMessages) {
				waitingIncoming.add(node);
//				log.info(Thread.currentThread().getName()+" add waitingIncoming: "+ node +
//						" recevedMessages size is: "+activeReceivedMessages.size()+
//						" and waitingIncoming is: "+waitingIncoming.size());
				activeReceivedMessages.wait(nonBlockingReceiveWaitTime);
				ret = scanForMessage();
				activeReceivedMessages.notify();
			}
		}
		
//		printOutQueues(true); //<- @FOR DEBUG
		
		/*
		 * It the waitingIncoming size increase too much, 
		 * probably the node
		 * is holding ancient messages.
		 * Otherwise this is perhaps the effect of the churn 
		 * and in that case the waitingIncoming queue
		 * may grow without any bound. 
		 */
		if(USE_PRUNING_LISTS_METHOD &&
				waitingIncoming.size() > BOUND_WAITINGINCOMING_SIZE){
			pruneWatingIncoming();
		}
		
		return ret;
	}

	/**
	 * Clean all the older than OLD_AGE_THRESHOLD 
	 * NodeDescriptr(s) that the queue is holding.
	 * 
	 * WARNING: this method may result in a bad behavior
	 * if a huge number of nodes having an age less then 
	 * OLD_AGE_THRESHOLD leaves contemporary the cloud.
	 * 
	 * WARNING: please be sure that the service uses the
	 * age of the NodeDescriptor or you'll incur in a 
	 * NullPointerException during the sort of the list.
	 *   
	 */
	private void pruneWatingIncoming() {
		// @FOR DEBUG
//		System.out.println("PRUNING: \n"+waitingIncoming );
		Collections.sort(waitingIncoming, new AgeInverseSorterer());
		while( (!waitingIncoming.isEmpty()) && 
				(((PssNodeDescriptor)waitingIncoming.firstElement()).getAge() >= OLD_AGE_THRESHOLD) ){
			waitingIncoming.remove(0);
		}
	}

	/**
	 * 
	 */
	@SuppressWarnings("unused")
	private void printOutQueues(boolean slim) {
		if(slim){
			System.out.println(this.getClass().getSimpleName()+" messages Queues:\n"+
					"active size: " + activeReceivedMessages.size()+"\n"+	
					"waiting size: "+ waitingIncoming.size()+"\n"+
					"passive size: " + passiveReceivedMessages.size()+"\n"
			);
		}else{
			System.out.println(this.getClass().getSimpleName()+" messages Queues:\n"+
					"active: "+ activeReceivedMessages+" size: "+activeReceivedMessages.size()+"\n"+	
					"waiting: "+waitingIncoming+" size: "+waitingIncoming.size()+"\n"+
					"passive: "+passiveReceivedMessages+" size: "+passiveReceivedMessages.size()+"\n"
			);
		}
	}

	@Override
	public Message receive() throws InterruptedException {
		Message ret = null;
		if(!passiveReceivedMessages.isEmpty()){
			ret = passiveReceivedMessages.remove(0);
		}
		while(ret == null){
			synchronized (passiveReceivedMessages) {
				passiveReceivedMessages.wait(0);
				if(!passiveReceivedMessages.isEmpty()){
					ret = passiveReceivedMessages.remove(0);
				}
				passiveReceivedMessages.notify();
			}
		}
		return ret;
	}
	
	/**
	 * @param node 
	 * @return
	 */
	protected Message scanForMessage(NodeDescriptor node) {
		Message ret = null;
		for(int i=0; ret == null && i<activeReceivedMessages.size(); i++){
			NodeDescriptor sender = activeReceivedMessages.get(i).getSender();
			if(sender.equals(node) && waitingIncoming.contains(sender)){
				ret = activeReceivedMessages.remove(i);
				waitingIncoming.remove(node);
			}
		}
		return ret;
	}
	
	/**
	 * Returns the first Message which the active thread
	 * is waiting for. For this purpose it checks 
	 * the waitingIncoming vector
	 * @return
	 */
	protected Message scanForMessage(){
		Message ret = null;
		for(int i=0; ret == null && i<activeReceivedMessages.size(); i++){
			Message m = activeReceivedMessages.get(i);
			NodeDescriptor sender = m.getSender();
			if(waitingIncoming.contains(sender)){
				ret = activeReceivedMessages.remove(i);
				waitingIncoming.remove(sender);
			}
		}
		return ret;
	}

	public static final int getBOUND_WAITINGINCOMING_SIZE() {
		return BOUND_WAITINGINCOMING_SIZE;
	}

	public static final int getOLD_AGE_THRESHOLD() {
		return OLD_AGE_THRESHOLD;
	}

	public static final boolean isUSE_PRUNING_LISTS_METHOD() {
		return USE_PRUNING_LISTS_METHOD;
	}

	public static final void setUSE_PRUNING_LISTS_METHOD(
			boolean uSEPRUNINGLISTSMETHOD) {
		USE_PRUNING_LISTS_METHOD = uSEPRUNINGLISTSMETHOD;
	}
}
